package peer

import (
	"errors"
	"time"

	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/p2p/common/constant"
	"hundsun.com/hsl/hschain/p2p/common/utils"
	pbp2p "hundsun.com/hsl/hschain/protos/p2p"
	"hundsun.com/hsl/hschain/store/mq"
)

func (p *Protocol) handleStreamPeerInfo(msg *pbp2p.P2PMsg) *pbp2p.P2PMsg {
	p.Log.Debugf("receive peerInfo request, peerID:%v", msg.PeerId)
	peerInfo := p.getLocalPeerInfo()
	buf, _ := proto.Marshal(peerInfo)
	resp := p.P2pNode.NewP2pMessage("", "", constant.MsgPeerInfo, buf, nil)
	return resp
}

func (p *Protocol) handleStreamConsensus(msg *pbp2p.P2PMsg) {
	p.Log.Infof("Receive consensus msg, send msg to module %s\n", msg.TargetMod)
	msgConsensus := p.TopicClient.NewMessage(msg.TargetMod, mq.ModeTopic.String(), types.EventConsensusRequestMsg, msg)
	err := p.TopicClient.Send(msgConsensus, false)
	if err != nil {
		p.Log.Errorf("send consensus message failed, reason:%v", err)
	}
}

func (p *Protocol) handleStreamConsensusSync(msg *pbp2p.P2PMsg) *pbp2p.P2PMsg {
	p.Log.Debugf("Receive consensus sync msg, send msg to module %s\n", msg.TargetMod)
	msgConsensus := p.WorkClient.NewMessage(msg.TargetMod, mq.ModeWorker.String(), types.EventConsensusSyncRequestMsg, msg)
	err := p.WorkClient.Send(msgConsensus, false)
	resp, err := p.WorkClient.WaitTimeout(msgConsensus, time.Second*4)
	if err != nil {
		p.Log.Errorf("send sync consensus message failed, reason:%v", err)
		return nil
	}
	oriMsg := resp.Data.(*pbp2p.P2PMsg)
	p2pMsg := p.P2pNode.NewConsensusP2pMessage("", "", constant.MsgConsensusSync, oriMsg.TargetPeerId, oriMsg.TargetMod, oriMsg.Payload, nil)
	id, err := utils.PeerIDDecode(oriMsg.TargetPeerId)
	if err != nil {
		p.Log.Errorf("consensus msg peer id illegal, peerID:%s, reason:%v", id, err)
		return nil
	}
	p.Log.Debugf("receive consensus sync msg, targetPeerID:%s, targetMod:%s", p2pMsg.TargetPeerId, p2pMsg.TargetMod)
	return p2pMsg
}

func (p *Protocol) handleEventPeerInfo(msg *mq.Message) {
	peers := p.P2pNode.PeerInfoMgr.FetchAll()
	p.Log.Debugf("receive p2p msg fetch peerInfo, num:%v", len(peers))
	msg.Reply(p.WorkClient.NewMessage("", "", types.EventPeerInfo, &pbp2p.PeerInfos{Peers: peers}))
}

func (p *Protocol) handleEventConsensusSync(msg *mq.Message) {
	p2pMsg, err := p.mq2P2p(msg)
	if err != nil {
		p.Log.Error(err)
		return
	}
	id, err := utils.PeerIDDecode(p2pMsg.TargetPeerId)
	p.Log.Debugf("receive consensus sync msg, targetPeerID:%s, targetMod:%s", p2pMsg.TargetPeerId, p2pMsg.TargetMod)
	//如果是本地，直接发送共识消息
	if id == p.P2pNode.Host.ID() {
		resp := p.handleStreamConsensusSync(p2pMsg)
		msg.Reply(p.WorkClient.NewMessage("", "", types.EventConsensusRequestMsg, resp))
		return
	}
	if err != nil {
		p.Log.Errorf("consensus msg peer id illegal, peerID:%s, reason:%v", id, err)
	}
	resp, err := p.P2pNode.SyncMsg(p2pMsg, id, constant.ConsensusSync)
	if err != nil {
		p.Log.Errorf("send sync p2p msg failed, reason:%v", err)
	}
	if resp == nil {
		return
	}
	msg.Reply(p.WorkClient.NewMessage("", "", types.EventConsensusRequestMsg, resp))
}

func (p *Protocol) handleEventConsensus(msg *mq.Message) {
	p2pMsg, err := p.mq2P2p(msg)
	if err != nil {
		p.Log.Error(err)
		return
	}
	id, err := utils.PeerIDDecode(p2pMsg.TargetPeerId)
	p.Log.Debugf("receive consensus msg, targetPeerID:%s, targetMod:%s", p2pMsg.TargetPeerId, p2pMsg.TargetMod)
	//如果是本地，直接发送共识消息
	if id == p.P2pNode.Host.ID() {
		p.handleStreamConsensus(p2pMsg)
		return
	}
	if err != nil {
		p.Log.Errorf("consensus msg peer id illegal, peerID:%s, reason:%v", id, err)
	}
	err = p.P2pNode.AsyncMsg(p2pMsg, id, constant.Consensus)
	if err != nil {
		p.Log.Errorf("send async p2p msg failed, reason:%v", err)
	}
}

func (p *Protocol) handleEventPeerID(msg *mq.Message) {
	p.Log.Debugf("receive p2p msg get peerID")
	pid := utils.PeerIDEncode(p.P2pNode.Host.ID())
	msg.Reply(p.WorkClient.NewMessage("", "", types.EventPeerID, pid))
}

func (p *Protocol) mq2P2p(msg *mq.Message) (*pbp2p.P2PMsg, error) {
	oriMsg := msg.Data.(*pbp2p.P2PMsg)
	if oriMsg == nil {
		return nil, errors.New("p2p msg not formatted")
	}
	p2pMsg := p.P2pNode.NewConsensusP2pMessage("", "", constant.MsgConsensus, oriMsg.TargetPeerId, oriMsg.TargetMod, oriMsg.Payload, nil)
	return p2pMsg, nil
}
